package midoo.server.core;

import io.netty.channel.ChannelHandlerContext;
import io.netty.util.*;
import io.netty.util.concurrent.DefaultThreadFactory;
import midoo.common.MConfig;
import midoo.common.MService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.*;

/**
 * Created by ajianzheng on 2015/4/17.
 *
 */
public class MMessageExecutor implements MService{
    protected static final Logger LOGGER = LoggerFactory.getLogger(MMessageExecutor.class);

    private static MMessageExecutor ourInstance = new MMessageExecutor();

    public static MMessageExecutor getInstance() {
        return ourInstance;
    }

    private MMessageExecutor() {
    }

    private static final Timer hashedWheelTimer = new HashedWheelTimer();

    private ExecutorService executor;
    private MMessageExecutorConfig config;

    @Override
    public int start(MConfig cfg) throws Exception {
        config = (MMessageExecutorConfig)cfg;

//        executor = Executors.newFixedThreadPool(config.getThreadPoolSize(), new DefaultThreadFactory("exec"));
        executor = new ThreadPoolExecutor(
                config.getThreadPoolSize(),     /// corePoolSize
                config.getThreadPoolSize(),     /// maximumPoolSize
                0,                              /// keepAliveTime
                TimeUnit.SECONDS,               /// unit
                new ArrayBlockingQueue<Runnable>(config.getWorkQueueSize()),   /// workQueue
                new DefaultThreadFactory("exec"),       /// threadFactory
                new ThreadPoolExecutor.AbortPolicy()    /// handler
                );

        return 0;
    }

    @Override
    public void stop() {
        executor.shutdown();
    }

    public void exec(final ChannelHandlerContext ctx,  Object message) {
        try {
            executor.submit(new Task(ctx, message));
        } catch (RejectedExecutionException e) {
            MServerBench.getInstance().handleOverload(MOverload.KOverloadType.MESSAGE, message);

            ReferenceCountUtil.release(message);
        }
    }

    private class Task implements Runnable {
        private Timeout timer;

        private ChannelHandlerContext ctx = null;
        private Object object = null;
        final Thread current = Thread.currentThread();

        public Task(final ChannelHandlerContext ctx,  Object object) {
            this.ctx = ctx;
            this.object = object;
        }

        @Override
        public void run() {
            MMessage message = new MMessage(ctx.channel(), object);

            /// 添加任务定时器
            timer = hashedWheelTimer.newTimeout(new TimerTask() {
                @Override
                public void run(Timeout timeout) throws Exception {
                    current.interrupt();
                }
            }, config.getTaskTimeout(), TimeUnit.MILLISECONDS);

            ///
            try {
                MServerBench.getInstance().handleProcess(message);
            }
            catch (Exception e) {     /// 任务超时，或者其他异常
                LOGGER.warn("task timeout,cost:{},remote:{},err:{}",
                        System.currentTimeMillis() - message.getProcessStartTimeMillis(),
                        message.getRemoteString(), e.getMessage());

                LOGGER.info("task timeout,msg:{},err:{}", message.getMessage(), e.getStackTrace());
            }
            finally {
                ReferenceCountUtil.release(object);

                if (null != timer) timer.cancel();
            }
        }
    }
}
